package org.wyz.spark;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class PageRank {

    private static final double d = 0.85;
    private static final String[] ARGS = {"data/input/pagerank/page.txt", "data/output/pagerank"};
    private static final int EPOCH = 5;

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("wyz-join");
        JavaSparkContext sc = new JavaSparkContext(conf);


        JavaRDD<String> file = sc.textFile(ARGS[0]);


        // 一共有几个节点
        long n = file.count();


        // 节点： PageRank值
        // 这个是要每轮更新的
        JavaPairRDD<String, Double> rank = file.mapToPair(a -> {
            String[] ss = a.split(" ");
            double pr = Double.parseDouble(ss[1]);
            return new Tuple2<>(ss[0], pr);
        });


        // 节点：[出节点]
        // 这个是固定的
        JavaPairRDD<String, List<String>> out = file.mapToPair(a -> {
            String[] ss = a.split(" ");
            List<String> res = new ArrayList<>();
            res.addAll(Arrays.asList(ss).subList(2, ss.length));
            return new Tuple2<>(ss[0], res);
        });

        for (int i = 1; i <= EPOCH; i++) {
            rank = rank.join(out)
                    .flatMap(a -> {
                        double pr = a._2._1;
                        List<String> end = a._2._2;
                        int degree = end.size();

                        List<Tuple2<String, Double>> res = new ArrayList<>();
                        for (String s : end) {
                            res.add(new Tuple2<>(s, pr / degree));
                        }
                        return res.iterator();
                    }).mapToPair(a -> new Tuple2<>(a._1, a._2))
                    .reduceByKey(Double::sum);
        }

        rank.map(a -> a._1 + " " + a._2).saveAsTextFile(ARGS[1]);
    }
}
